package com.leilei.cdc;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lei
 * @create 2022-01-12 17:52
 * @desc
 **/
public class FlinkCdcMain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
                .hostname("10.50.50.196")
                .port(3306)
                .databaseList("paas_common_db_alpha")
                .tableList("paas_common_db_alpha.base_business_driver_score")
                .username("root")
                .password("123456")
                /**initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                .startupOptions(StartupOptions.initial())
                // .deserializer(new StringDebeziumDeserializationSchema())
                .deserializer(new CustomerDeserializationSchema())
                .build();
        // 4、使用CDC Source方式从mysql中读取数
        final DataStreamSource<String> streamSource = env.addSource(mysqlSource);
        streamSource.print();
        env.execute("aa");

    }
}
